package com.example.case2;

import org.apache.zookeeper.*;
import org.apache.zookeeper.data.Stat;

import java.io.IOException;
import java.util.Collections;
import java.util.List;
import java.util.concurrent.CountDownLatch;

/**
 * @ClassName DistributeLock
 * @Author 小坏
 * @Date 2021/8/22、19:43
 * @Version 1.0
 */
public class DistributeLock {

    private String connectString = "hadoop102:2181,hadoop103:2181,hadoop104:2181";
    private int sessionTimeout = 2000;
    private ZooKeeper zooKeeper;

    private String waitPath;

    private CountDownLatch downLatch = new CountDownLatch(1);

    private CountDownLatch waith = new CountDownLatch(1);
    private String currentMode;

    public DistributeLock() throws IOException, InterruptedException, KeeperException {

        //1、获取连接
        zooKeeper = new ZooKeeper(connectString, sessionTimeout, new Watcher() {
            @Override
            public void process(WatchedEvent watchedEvent) {

                //connectLath 如果连接上zk 可以释放
                if (watchedEvent.getState() == Event.KeeperState.SyncConnected) {
                    downLatch.countDown();
                }


                //waitLath 如果连接上zk 可以释放
                if (watchedEvent.getType() == Event.EventType.NodeDeleted && watchedEvent.getPath()   .equals(waitPath)) {
                    waith.countDown();
                }

            }
        });

        //等待 zk正常连接后，往下再走程序
        downLatch.await();
        Stat exists = zooKeeper.exists("/locks", false);
        if (exists == null) {
            //不存在就创建locks 节点
            zooKeeper.create("/locks", "locks".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE
                    , CreateMode.PERSISTENT);
        }

        //盘那段根节点 /locks是否存在

    }

    //对zk加锁
    public void zkLock() throws KeeperException, InterruptedException {
        //创建临时带序列化号的节点
        currentMode = zooKeeper.create("/locks/" + "seq-", null,
                ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL);
        //判断创建的节点是否是最小的序号节点，如果是 获取到锁; 如果不是，监听他的序列号前一个节点

        List<String> children = zooKeeper.getChildren("/locks", false);
        //如果children 只有一个值，那就直接获取锁、如果有多个节点、需要判断、谁最小
        if (children.size() == 1) {
            return;
        } else {
            Collections.sort(children);

            //获取节点名称、把locks 戒掉了-(出来的数据是 seq-0000000000000000)
            String thisNode = currentMode.substring("/locks/".length());
            //通过seq-0000000000000000 获取该节点在 children 集合的位置

            int index = children.indexOf(thisNode);

            if (index == -1) {
                System.out.println("数据异常！");
            } else if (index == 0) {
                //就一个节点、可以获取锁了
                return;
            } else {
                //如果他不是最后一个节点、需要监听  他前一个节点变化

                waitPath = "/locks/" + children.get(index - 1);
                zooKeeper.getData(waitPath, true, null);

                //等待结束
                waith.await();
                return;
            }
        }
    }


    //解锁
    public void unLock() throws KeeperException, InterruptedException {
        zooKeeper.delete(currentMode, -1);
    }


}
